package com.wjc.train.business.mq;

import com.alibaba.fastjson2.JSON;
import com.wjc.train.business.enums.ConfirmOrderStatusEnum;
import com.wjc.train.business.req.ConfirmOrderSaveReq;
import com.wjc.train.business.service.IConfirmOrderService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.MDC;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;

/**
 * @author Guessed
 */
@Service
@Slf4j
@RocketMQMessageListener(consumerGroup = "default",topic = "CONFIRM_ORDER")
public class ConfirmOrderConsumer implements RocketMQListener<MessageExt> {

    @Resource
    private IConfirmOrderService confirmOrderService;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void onMessage(MessageExt messageExt) {
        //考虑幂等,获取消息唯一ID
        String messageKey = messageExt.getProperty("key");
        //如果已经消费过那么直接return
        byte[] body = messageExt.getBody();
        ConfirmOrderSaveReq saveReq = JSON.parseObject(new String(body), ConfirmOrderSaveReq.class);
        MDC.put("LOG_ID", saveReq.getLogId());
        log.info("消息key:{}",messageKey);
        //用redis做幂等,消息ID作为key,value为消息状态
        //初始进来的消息置为I(初始状态)
        //消费中的置为P
        //消费成功的置为S
        //消费失败的直接删除
        Boolean ifAbsent =
                stringRedisTemplate.opsForValue().setIfAbsent(messageKey, ConfirmOrderStatusEnum.PENDING.getCode(), Duration.ofMinutes(30L));

        if(Boolean.FALSE.equals(ifAbsent)){
            String status = stringRedisTemplate.opsForValue().get(messageKey);
            if(ConfirmOrderStatusEnum.SUCCESS.getCode().equals(status) || ConfirmOrderStatusEnum.PENDING.getCode().equals(status)){
                return;
            }
        }

        confirmOrderService.doConfirm(saveReq,messageKey);
        log.info("消费者收到消息:{}",new String(body));
    }
}
